package com.gagakuai.bpm.mqlistener.camunda.baseproxy;




import com.gagakuai.bpm.common.component.SpringContextHolder;
import com.gagakuai.bpm.common.constants.BpmMetaConstants;
import com.gagakuai.bpm.common.constants.MsgLogConstants;
import com.gagakuai.bpm.domian.wapper.DeadMessageWrapper;
import com.gagakuai.bpm.utils.ObjectUtils;
import com.gagakuai.bpm.utils.RedisUtil;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedDeque;

/*
 * @program: mss_server
 * @description: 消费者代理类
 * @author: houhong
 * @create: 2022-03-24 19:44
 */
@Slf4j
public class BaseConsumerProxy implements InvocationHandler {

    /*
     * 被代理的类
     */
    private Object target;


    private RedisUtil redisUtil;


    public BaseConsumerProxy(Object target, RedisUtil redisUtil) {
        this.target = target;
        this.redisUtil = redisUtil;
    }

    public Object getProxy() {

        log.info("#===============进入消息代理,进行消息重复消费检测,消息消费失败后重新入队防止消息丢失保障============#");
        ClassLoader classLoader = this.target.getClass().getClassLoader();

        Class<?>[] interfaces = this.target.getClass().getInterfaces();

        Object proxy = Proxy.newProxyInstance(classLoader, interfaces, this);
        return proxy;

    }

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {


        Message message = null;
        Channel channel = null;
        String messageType = null;
        String correlationId = null;
        Map<Object, Object> hmget = null;

        Long tag = null;

        String key = null;

        //直接将数据存入queue
        DeadMessageWrapper deadMessageWrapper = new DeadMessageWrapper();

        try {
            message = (Message) args[0];
            channel = (Channel) args[1];
            messageType = (String) args[2];
            correlationId = getCorrelationId(message);


            deadMessageWrapper.setType(messageType);
            deadMessageWrapper.setMessage(message);
            deadMessageWrapper.setCorrelationId(correlationId);
            deadMessageWrapper.setChannel(channel);

            String deadMessageKey = BpmMetaConstants.BPM_DEAD_MSG_TYPE
                    + BpmMetaConstants.REDIS_FILE_KEY_SEPTOR + correlationId;

            //todo 考虑使用SETNX 原语
        /*if (!redisUtil.hasKey(deadMessageKey)) {
            redisUtil.set(deadMessageKey, jsonString);
        }*/

            //这里需要使用CAS吗？ 保证并发
            key = MsgLogConstants.MSG_INFO_BPM_PREX_MSS + correlationId;
            if (isConsumed(key)) {
                log.info("重复消费, correlationId: {}", correlationId);
                return null;
            }

            MessageProperties properties = message.getMessageProperties();
            tag = properties.getDeliveryTag();




            /*
             *
             * 真正消费的业务逻辑
             */
            Object result = method.invoke(target, args);
            //确认消息已经被消费
            //msgLogApi.changeBrokerMessageLogStatus(correlationId, MsgLogConstants.BPM_WITH_DEAW_CONSUME_SUCCESS);
            hmget = redisUtil.hmget(key);

            hmget.put("status", MsgLogConstants.BPM_WITH_DEAW_CONSUME_SUCCESS);
            //信息存放7天 -- 或着删除消息 ? todo
            redisUtil.hmsetObj(key, hmget, 60 * 60 * 7);
            // 消费确认
            channel.basicAck(tag, false);
            //回删redis
            redisUtil.del(deadMessageKey);
            //回删 系统重试标志key
            String retryKey = BpmMetaConstants.BPM_DEAD_MSG_SYSTEM_RETRY
                    + BpmMetaConstants.REDIS_FILE_KEY_SEPTOR
                    + correlationId;
            redisUtil.del(retryKey);
            log.info("#===============消息消费成功:correlationId:{}============#", correlationId);
            return result;
        } catch (Exception e) {

            log.error("获取代理失败", e);

            hmget = redisUtil.hmget(key);
            //设置消息重试次数-->消费者
            Object comsumerRetryCountObj = hmget.get("comsumerRetryCount");


            Integer comsumerRetryCount = new Integer(0);
            if (ObjectUtils.isNull(comsumerRetryCountObj)) {
                comsumerRetryCount = 0;
            } else {
                if (comsumerRetryCountObj instanceof Integer) {

                    comsumerRetryCount = (Integer) comsumerRetryCountObj;
                } else if (comsumerRetryCountObj instanceof String) {
                    String comsumerRetryCountStr = (String) comsumerRetryCountObj;
                    comsumerRetryCount = Integer.parseInt(comsumerRetryCountStr);
                }

            }
            comsumerRetryCount++;
            if (comsumerRetryCount > MsgLogConstants.CONSUMER_RETRY_THREAD_SHOLD) {
                //放入到死信队列里面
                log.info("#=====消费端达到最大消费重试值,转入死信队列【消息id】:{}", hmget.get("messageId"));
                hmget.put("comsumerRetryCount", comsumerRetryCount);
                hmget.put("status", MsgLogConstants.BPM_WITH_DEAW_CONSUME_FAILURE);
                redisUtil.hmset(key, hmget);
                //todo 将进入死信的消息放入queue,在死信监听器里面使用
                //回删 系统重试标志key
                String retryKey = BpmMetaConstants.BPM_DEAD_MSG_SYSTEM_RETRY
                        + BpmMetaConstants.REDIS_FILE_KEY_SEPTOR
                        + correlationId;
                if (!redisUtil.hasKey(retryKey)) {
                    Queue<DeadMessageWrapper> localQueue = SpringContextHolder.getBean(BpmMetaConstants.DEAD_LOCAL_QUEUE_NAME,
                            ConcurrentLinkedDeque.class);
                    localQueue.add(deadMessageWrapper);
                }
                //todo 做业务切分？
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
            } else {
                // 消息重新入队
                //// TODO: 2022/6/28  在消息消费失败了一定次数后是不是需要进行转入死信队列？
                hmget.put("comsumerRetryCount", comsumerRetryCount + "");
                redisUtil.hmset(key, hmget);
                channel.basicNack(tag, false, true);
                log.info("#===============消息重新入队:correlationId:{},tag:{}============#", correlationId, tag);
            }
            return null;
        }

    }


    /*
     * 获取CorrelationId
     *
     * @param message
     * @return
     */
    private String getCorrelationId(Message message) {

        String correlationId = null;

        MessageProperties properties = message.getMessageProperties();
        Map<String, Object> headers = properties.getHeaders();
        for (Map.Entry entry : headers.entrySet()) {
            String key = (String) entry.getKey();
            if ("spring_returned_message_correlation".equals(key)) {
                String value = (String) entry.getValue();
                correlationId = value;
                return correlationId;
            }
        }

        return null;
    }

    /*
     * 消息是否已被消费
     *
     * @param correlationId
     * @return
     */
    private boolean isConsumed(String correlationId) {

        Map<Object, Object> msgLogMap = redisUtil.hmget(correlationId);


        if (null == msgLogMap || msgLogMap.get("status").equals(MsgLogConstants.BPM_WITH_DEAW_CONSUME_SUCCESS)) {
            return true;
        }
        return false;
    }
}